package cn.itcast.kafka;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;

public class FlinkKafkaProducer {
    public static void main(String[] args) throws Exception {
        /**
         * 1.创建运行环境
         * 2.设置按照事件时间处理数据
         * 3.读取数据
         * 4.将数据写入到kafka中
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStreamSource<String> lines = env.readTextFile("data/sourcedata.txt");

        FlinkKafkaProducer011 flinkKafkaProducer011 = new FlinkKafkaProducer011(
                "node01:9092,node02:9092,node03:9092",
                "vehicledata-dev",new SimpleStringSchema()
        );

        lines.addSink(flinkKafkaProducer011);

        env.execute();


    }
}
